package com.boyi.center.task;

import com.boyi.shared.kafka.KafkaResult;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.stereotype.Service;

/**
 * @author tingwen
 */
@Service
public class DefaultKafkaTask extends AbstractKafkaTask {

	@Override
	public void handleSuccess(Object result) {
		System.out.println(" success ");
	}

	@Override
	public void handleFailure(ProducerRecord<String, KafkaResult<Object>> record) {
		KafkaResult<Object> value = record.value();
		int notifyTime = value.getNotifyTime();
		if (notifyTime < FAILURE_TIME) {
			value.setNotifyTime(++notifyTime);
			ProducerRecord<String, KafkaResult<Object>> newRecord = new ProducerRecord<>(record.topic(), value);
			taskPublic(newRecord);
		} else {
			logger.debug("notifyTime 超过" + FAILURE_TIME + "次");
		}
	}
}
